package com.niit.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable

/**
 * Date:2025/5/21
 * Author：Ys
 * Description:
 */
object SparkStreaming03_Transform {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming03_Transform")
    val ssc = new StreamingContext(conf,Seconds(3))
    ssc.sparkContext.setLogLevel("ERROR")

    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

    //transform:DStream功能不完整的时候，或者DStream没有相应的操作，那么可以使用transform
    val wordDS: DStream[(String, Int)] = lines.transform(rdd => {
      val value: RDD[(String, Int)] = rdd.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey(_ + _)
      value

    })

    wordDS.print()

    ssc.start()

    ssc.awaitTermination()
  }

}
